#include <assert.h>
#include "work_queue.h"
 
typedef struct _THREAD_CONTEXT {
   WorkQueue* work_queue_;
   void* thread_data_;
} THREAD_CONTEXT, *PTHREAD_CONTEXT;
 
bool WorkQueue::create(const unsigned int  num_of_threads, void** thread_data/*=NULL*/) {
   work_item_queue_ = new WorkItemQueue();
   if(NULL == work_item_queue_) {
	  return false;
   }
   sync_obj_array_[SEMAPHORE_INDEX] = CreateSemaphore(NULL, 0, LONG_MAX, NULL);
   if(sync_obj_array_[SEMAPHORE_INDEX] == NULL) {
	   delete work_item_queue_;
	   work_item_queue_ = NULL;
	   return false;
   }
   sync_obj_array_[ABORT_EVENT_INDEX] = CreateEvent(NULL, TRUE, FALSE, NULL);
   if (sync_obj_array_[ABORT_EVENT_INDEX] == NULL) {
	   delete work_item_queue_;
	   work_item_queue_ = NULL;
	   CloseHandle(sync_obj_array_[SEMAPHORE_INDEX]);
	   return false;
   }
   InitializeCriticalSection(&cs_);
   threads_ = new HANDLE[num_of_threads];
   if (threads_ == NULL) {
	   delete work_item_queue_;
	   work_item_queue_ = NULL;
	   CloseHandle(sync_obj_array_[SEMAPHORE_INDEX]);
	   CloseHandle(sync_obj_array_[ABORT_EVENT_INDEX]);
	   DeleteCriticalSection(&cs_);
	   return false;
   }
   num_of_threads_ = num_of_threads;
   DWORD thread_id;
   PTHREAD_CONTEXT thread_context;
   for(unsigned int i = 0 ; i < num_of_threads ; i++ ) {
	  thread_context = new THREAD_CONTEXT;
	  thread_context->work_queue_  = this;
	  thread_context->thread_data_ = thread_data == NULL? NULL : thread_data[i];
	  threads_[i] = CreateThread(NULL,
		  0,
		  WorkQueue::thread_func,
		  thread_context,
		  0,
		  &thread_id
	  );
	  if(threads_[i] == NULL) {
		 delete thread_context;
		 num_of_threads_ = i;
		 destroy(5);
		 return false;
	  }
   }
   return true;
}
 
bool WorkQueue::insert_work_item(WorkItemBase* work_iterm) {
	assert(work_iterm != NULL);
	EnterCriticalSection(&cs_);
	work_item_queue_->push(work_iterm);
	LeaveCriticalSection(&cs_);
	if (!ReleaseSemaphore(sync_obj_array_[SEMAPHORE_INDEX], 1, NULL)) {
		assert(false);
		return false;
	}
	return true;
}
 
WorkItemBase* WorkQueue::remove_work_item() {
	WorkItemBase* work_item;
	EnterCriticalSection(&cs_);
	work_item = work_item_queue_->front();
	work_item_queue_->pop();
	LeaveCriticalSection(&cs_);
	assert(work_item != NULL);
	return work_item;
}

unsigned long __stdcall WorkQueue::thread_func(void* param) {
	PTHREAD_CONTEXT thread_context = (PTHREAD_CONTEXT)param;
	WorkItemBase* work_item = NULL;
	WorkQueue* work_queue = thread_context->work_queue_;
	void* thread_data = thread_context->thread_data_;
	DWORD wait_result;
	for (;;) {
		wait_result = WaitForMultipleObjects(NUMBER_OF_SYNC_OBJ, work_queue->sync_obj_array_, FALSE, INFINITE);
		switch (wait_result - WAIT_OBJECT_0) {
		case ABORT_EVENT_INDEX:
			delete thread_context;
			return 0;
		case SEMAPHORE_INDEX:
			work_item = work_queue->remove_work_item();
			if (work_item == NULL) {
				assert(false);
				break;
			}
			work_item->work(thread_data);
			break;
		default:
			assert(false);
			delete thread_context;
			return 0;
		}
	}
	delete thread_context;
	return 1;
}
 
int WorkQueue::get_threads_num() {
	return num_of_threads_;
}

void WorkQueue::suspend() {
	DWORD code = 0;
	for (unsigned int i = 0; i < num_of_threads_; i++) {
		GetExitCodeThread(threads_[i], &code);
		if (code == STILL_ACTIVE) {
			SuspendThread(threads_[i]);
		}
	}
}

void WorkQueue::resume() {
	DWORD code = 0;
	for (unsigned int i = 0; i < num_of_threads_; i++) {
		GetExitCodeThread(threads_[i], &code);
		if (code == STILL_ACTIVE) {
			ResumeThread(threads_[i]);
		}
	}
}
 
size_t WorkQueue::get_work_queue_size() {
	EnterCriticalSection(&cs_);
	size_t size = work_item_queue_->size();
	LeaveCriticalSection(&cs_);
	return size;
}
 
void WorkQueue::destroy(int wait_seconds) {
	if (threads_ == nullptr) {
		return;
	}
	while (0 != get_work_queue_size()) {
		Sleep(wait_seconds * 1000);
	}

	if (!SetEvent(sync_obj_array_[ABORT_EVENT_INDEX])) {
		assert(false);
		return;
	}
	WaitForMultipleObjects(num_of_threads_, threads_, true, INFINITE);
	while (!work_item_queue_->empty()) {
		work_item_queue_->front()->abort();
		work_item_queue_->pop();
	}
	delete work_item_queue_;
	work_item_queue_ = NULL;
	CloseHandle(sync_obj_array_[SEMAPHORE_INDEX]);
	CloseHandle(sync_obj_array_[ABORT_EVENT_INDEX]);
	DeleteCriticalSection(&cs_);

	for (unsigned int i = 0; i < num_of_threads_; i++)
		CloseHandle(threads_[i]);

	delete[] threads_;
}